/*
 * Copyright (c) 2018-2023 北京华恒盛世科技有限公司
 * QUICKPOOL is licensed under Mulan PSL v2.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 *     http://license.coscl.org.cn/MulanPSL2
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
*/

#include "ee.h"

#define CHILD_DELETED     2

static int exec_command(struct executeCommand *);

extern char **environ;

/**
 * @brief       To execute job.
 *
 * @param[in]   jargv        #1: job argument
 *
 * @note        We need block most of signals before executing the job.
 */
void
ee_executejob(char **jargv)
{
    sigset_t sigMask, oldSigMask;
    
    sigemptyset(&sigMask);
    sigaddset(&sigMask, SIGCHLD);
    sigaddset(&sigMask, SIGINT);
    sigaddset(&sigMask, SIGHUP);
    sigaddset(&sigMask, SIGPIPE);
    sigaddset(&sigMask, SIGTTIN);
    sigaddset(&sigMask, SIGTTOU);
    sigaddset(&sigMask, SIGTSTP);
#ifdef SIGDANGER
    sigaddset(&sigMask, SIGDANGER);
#endif
    sigaddset(&sigMask, SIGTERM);
#ifdef SIGXCPU
    sigaddset(&sigMask, SIGXCPU);
#endif
#ifdef SIGXFSZ
    sigaddset(&sigMask, SIGXFSZ);
#endif
#ifdef SIGPROF
    sigaddset(&sigMask, SIGPROF);
#endif
#ifdef SIGLOST
    sigaddset(&sigMask, SIGLOST);
#endif
    sigaddset(&sigMask, SIGUSR1);
    sigaddset(&sigMask, SIGUSR2);
#ifdef SIGABRT
    sigaddset(&sigMask, SIGABRT);
#endif
    sigprocmask(SIG_BLOCK, &sigMask, &oldSigMask);

    struct executeCommand cmdBill;
    memset(&cmdBill, 0, sizeof(struct executeCommand));

    cmdBill.options = 0;

    char *pwd;
    if ((pwd = getenv("PWD"))) {
        strncpy(cmdBill.cwd, pwd, PATH_MAX);
    } else {
        cmdBill.cwd[0] = '\0';
    }

    cmdBill.argv = jargv;

    int ret;
    ret = exec_command(&cmdBill);
    if (0 > ret) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL, __func__, "exec_command");
        ee_quit(99);
    }

    child_ee = TRUE;
    child_go = TRUE;

    sigprocmask(SIG_SETMASK, &oldSigMask, NULL);

    return;
} // end function ee_executejob

/**
 * @brief       send signal to my task.
 *
 * @param[in]   sig          #1: signal to be sent
 *
 * @retval      0            #1: succeed
 * @retval      -1           #2: failed, process has been gone
 */
int
kill_mytask(int sig)
{
    if (0 == gptrTasks[0].pgid) {
        if (0 == gptrTasks[0].pid) {
            return 0;
        }

	if (kill(gptrTasks[0].pid, sig) == 0) {
            return (kill(-gptrTasks[0].pid, sig));
	}
    } else {
        return (kill(-gptrTasks[0].pgid, sig));
    }

    return 0;
} // end function kill_mytask

/* ------------------------------------------------
 *
 *  below are static functions used in this file
 *
 * ------------------------------------------------ */

static int
pairsocket(int af, int type, int protocol, int *sv)
{
    struct sockaddr_in   sa;
    struct hostent       *hp;
    int                  s;
    socklen_t            sa_size;
    fd_set               fdReadMask;
    struct timeval       tvTime;
    int                  iRetVal;

#define PAIRSOCKET_TIMEOUT      30

    if (debug > 1)
        printf("pairsocket\n");

    if (af != AF_INET || type != SOCK_STREAM)
        return -1;

    if ((hp = gf_gethostbyname(gstrMyName)) == NULL) {
        log_message(ERR, NOCLASS, "%s: gf_gethostbyname() failed for host %s", __func__, gstrMyName);
        return -1;
    }

    if ((s = socket(af, type, protocol)) < 0) {
        log_message(ERR, NOCLASS, STR_FUNC_S_FAIL_M, __func__, "malloc", "af");
        return -1;
    }

    memset((char*)&sa, 0, sizeof(sa));
    sa.sin_addr.s_addr = INADDR_ANY;
    sa.sin_family = AF_INET;
    sa.sin_port = 0;

    if (gf_sock_bind(s, (struct sockaddr *)&sa, sizeof(sa)) < 0) {
        log_message(ERR, NOCLASS, STR_FUNC_FAIL_M, __func__, "gf_sock_bind");
        close(s);
        return -1;
    }

    sa_size = sizeof(sa);
    if (getsockname(s, (struct sockaddr *)&sa, &sa_size) < 0) {
        log_message(ERR, NOCLASS, STR_FUNC_FAIL_M, __func__, "getsockname");
        close(s);
        return -1;
    }

    if (listen(s,1) < 0) {
        log_message(ERR, NOCLASS, STR_FUNC_FAIL_M, __func__, "listen");
        close(s);
        return -1;
    }

    if ((sv[1]=socket(af, type, protocol)) < 0) {
        log_message(ERR, NOCLASS, STR_FUNC_FAIL_M, __func__, "socket");
        close(s);
        return -1;
    }
    memcpy((char *)&sa.sin_addr, hp->h_addr_list[0], hp->h_length);
 retry:
    if (connect(sv[1], (struct sockaddr *)&sa, sizeof(sa)) < 0 ) {
        int err;
        err = errno;

        log_message(ERR, NOCLASS, STR_FUNC_FAIL_M, __func__, "connect");

        if (err == EINTR) {
            goto retry;
	}

        if (err != EINPROGRESS) {
            close(s);
            close(sv[1]);
            return (-1);
        }

        close(s);
        close(sv[1]);
        return -1;
    }

    log_message(DEBUG, EXECUTE, "%s: connected to passive socket '%d'", __func__, s);

    FD_ZERO(&fdReadMask);
    FD_SET(s, &fdReadMask);

    tvTime.tv_sec = PAIRSOCKET_TIMEOUT;
    tvTime.tv_usec = 0;

    log_message(DEBUG, EXECUTE, "%s: begin select(), timeout in '%d' sec.",
                __func__, (int)tvTime.tv_sec);

    iRetVal = select(s+1, &fdReadMask, NULL, NULL, &tvTime);
    if (iRetVal == 0) {
        log_message(DEBUG, EXECUTE, "%s: select() timed out.", __func__);

        close(s);
        close(sv[1]);
        return -1;
    } else if (iRetVal == -1) {

        log_message(DEBUG, EXECUTE, "%s: select() failed: %m", __func__);
        close(s);
        close(sv[1]);
        return -1;
    } else if (FD_ISSET(s, &fdReadMask)) {
        log_message(DEBUG, EXECUTE, "%s: select() ok", __func__);

        sa_size = sizeof(sa);

        if ((sv[0] = accept(s, (struct sockaddr *)&sa, &sa_size)) < 0) {
            log_message(DEBUG, EXECUTE, STR_FUNC_FAIL_M, __func__, "accept");
            close(s);
            close(sv[1]);
            return -1;
        }

        close(s);
        return 0;
    }

    log_message(DEBUG, EXECUTE, "%s: unknown return from select, ret=%d", __func__, iRetVal);

    close(s);
    close(sv[1]);

    return -1;
}

static int
exec_command(struct executeCommand *cmdmsg)
{
    int sv[2], errSock[2];

    pid_t pid;
    if (pairsocket(AF_INET, SOCK_STREAM, 0, sv) < 0) {
        log_message(ERR, NOCLASS, STR_FUNC_FAIL_M, __func__, "pairsocket");
        return -1;
    }

    if (pairsocket(AF_INET, SOCK_STREAM, 0, errSock) < 0) {
        log_message(ERR, NOCLASS, STR_FUNC_FAIL_M, __func__, "pairsocket");
        close(sv[0]);
        close(sv[1]);
        return -1;
    }

    pid = fork();

    if (0 > pid) {

        log_message(ERR, NOCLASS, STR_FUNC_FAIL_M, __func__, "fork");

        close(sv[0]);
        close(sv[1]);
        close(errSock[0]);
        close(errSock[1]);

        return (pid);
    }

    if (0 < pid) {

        close(sv[1]);
        gf_set_nonblocking(sv[0]);

        close(errSock[1]);
        gf_set_nonblocking(errSock[0]);
    } else {

        close(sv[0]);
        close(errSock[0]);

        if (gboolJobMode || gboolTaskMode) {
            if (NULL != gstrMyName) {
                gf_setenv("HOSTNAME", gstrMyName);
            }

            gf_chan_close_accept();

            sigset_t sigMask;
            int i, iofd;
            struct linger linger;

            iofd = sv[1];
            linger.l_onoff = 1;
            linger.l_linger = 60;

            if (setsockopt(iofd, SOL_SOCKET, SO_LINGER, (char *)&linger,
                           sizeof(struct linger)) < 0) {
                log_message(ERR, NOCLASS, "%s: setsockopt on socket=<%d> option SO_LINGER failed: %m",
                            __func__, iofd);
            }

            pid_t myPid = getpid();
            if (setpgid(0, myPid) < 0) {
                log_message(ERR, NOCLASS, "%s: setpgid() failed due to %m.", __func__);
            }

            if (cmdmsg->options & REMOTE_STDERR) {
                if (setsockopt(errSock[1], SOL_SOCKET, SO_LINGER, (char *)&linger,
                               sizeof(struct linger)) < 0) {
                    log_message(ERR, NOCLASS, "%s: setsockopt on socket=<%d> option SO_LINGER failed: %m",
                                __func__, errSock[1]);
                }
            }

            if (!(gintExecFlag & JM_FLAG_STDIN)) {
                dup2(iofd, 0);
            }

            if (!(gintExecFlag & JM_FLAG_STDOUT)) {
                dup2(iofd, 1);
            }

            if (!(gintExecFlag & JM_FLAG_STDERR)) {

                if (cmdmsg->options & REMOTE_STDERR) {
                    dup2(errSock[1], 2);
                } else {
                    dup2(iofd, 2);
                }
            }

            for (i = 1; i < NSIG; i++)
                Signal_(i, SIG_DFL);

            Signal_(SIGHUP, SIG_IGN);

            sigemptyset(&sigMask);
            sigprocmask(SIG_SETMASK, &sigMask, NULL);

            close_log();

            execvp(cmdmsg->argv[0], cmdmsg->argv);
            perror(cmdmsg->argv[0]);
        }

        exit(127);
    }

    gptrTasks[0].inst_usage = (struct jobResourceUsage *)calloc(1, sizeof(struct jobResourceUsage));
    if (NULL == gptrTasks[0].inst_usage) {

        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc",
                    sizeof(struct jobResourceUsage));

        closesocket(sv[0]);
        closesocket(sv[1]);
        closesocket(errSock[0]);
        closesocket(errSock[1]);

        return -1;
    }

    gptrTasks[0].pid = pid;
    gptrTasks[0].pgid = pid;

    gptrTasks[0].exec_option = cmdmsg->options;

    if (NULL != cmdmsg->cwd)
        gptrTasks[0].cwd = copy_string(cmdmsg->cwd);
    else {
        gptrTasks[0].cwd = NULL;
        log_message(ERR, NOCLASS, "%s: cwd is null", __func__);
    }

    gptrTasks[0].start_time = time(NULL);

    log_message(DEBUG, EXECUTE, "%s: Child with pid=<%d> executed", __func__, pid);
    dump_task(&(gptrTasks[0]), 0, "first node task");

    if (debug > 1) {
        printf("child(%d)'s task_number=%d\n", gptrTasks[0].pid, gintNumTask);
        fflush(stdout);
    }

    return 0;
}

void
dump_task(struct task *child, int rank, const char *why)
{
    log_message(DEBUG, EXECUTE, "%s: %s child=<%x> node_rank=<%d> child->pid=<%d> exec_option=<%d>.",
                __func__, why, child, rank, child->pid, child->exec_option);

    if (NULL != child->inst_usage) {
        log_message(DEBUG, EXECUTE, "%s: mem=%ld, swap=%ld, cputime=%d, pid=%d, pgid=%d, npid=%d.",
                    __func__, child->inst_usage->mem, child->inst_usage->swap,
                    (child->inst_usage->utime+child->inst_usage->stime),
                    child->pid, child->pgid, child->inst_usage->npids);
    }

    return;
}
